package cn.monkey.io.transport.mq.rabbitmq;

import cn.monkey.io.transport.mq.ConcurrentMapExchangeContext;
import cn.monkey.io.transport.core.Exchange;
import cn.monkey.io.transport.core.ExchangeContext;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.concurrent.atomic.AtomicBoolean;

public class SimpleRabbitMqExchange implements RabbitMqExchange {
    private final String id;
    private final RabbitMqInbound rabbitMqInbound;
    private final RabbitMqOutbound rabbitMqOutbound;
    private final AtomicBoolean isActive;
    private final ExchangeContext context;

    public SimpleRabbitMqExchange(String id,
                                  RabbitMqInbound rabbitMqInbound,
                                  RabbitMqOutbound rabbitMqOutbound) {
        this.id = id;
        this.isActive = new AtomicBoolean(true);
        this.rabbitMqInbound = rabbitMqInbound;
        this.rabbitMqOutbound = rabbitMqOutbound;
        this.context = new ConcurrentMapExchangeContext();
    }

    @Override
    public String id() {
        return this.id;
    }

    @Override
    public ExchangeContext context() {
        return this.context;
    }

    @Override
    public Mono<Void> close() {
        if (this.isActive.compareAndSet(false, true)) {
            return Mono.error(new IllegalStateException("exchange has already closed"));
        }
        Mono<Void> close = this.inbound().close();
        Mono<Void> close1 = this.outbound().close();
        return Flux.merge(close, close1).then();
    }

    @Override
    public boolean isActive() {
        return this.isActive.get();
    }

    @Override
    public RabbitMqExchange bind() {
        this.setAttribute(Exchange.class.getName() + "#" + Exchange.KEY, this);
        return this;
    }

    @Override
    public RabbitMqInbound inbound() {
        return this.rabbitMqInbound;
    }

    @Override
    public RabbitMqOutbound outbound() {
        return this.rabbitMqOutbound;
    }

}
